package org.example.update_table;

import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class PaimonInsertOverwriteThenUpdate {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.setString(RestOptions.BIND_PORT,"8081");
        
        // 创建批处理模式的TableEnvironment
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inBatchMode()  // 明确指定为批处理模式
                .build();
        TableEnvironment tableEnv = TableEnvironment.create(settings);
        
        // 打印SQL语句
        String createCatalogSql = "CREATE CATALOG paimon WITH (" +
                "    'type' = 'paimon'," +
                "    'warehouse' = 'file:///tmp/paimon'" +
                ");";
        System.out.println("执行SQL语句：\n" + createCatalogSql);
        
        // 创建Paimon catalog
        tableEnv.executeSql(createCatalogSql);

        // 打印SQL语句
        String dropTableSql = "DROP TABLE IF EXISTS `paimon`.`default`.`user_behavior_copy`";
        System.out.println("执行SQL语句：\n" + dropTableSql);

        tableEnv.executeSql(dropTableSql);

        // 打印SQL语句 - 创建支持更新的Paimon表
        String createTableSql = "CREATE TABLE IF NOT EXISTS `paimon`.`default`.`user_behavior_copy` (\n" +
                "    user_id STRING,\n" +
                "    item_id STRING,\n" +
                "    behavior BIGINT,\n" +
                "    dt STRING,\n" +
                "    eventtime TIMESTAMP_LTZ(3),\n" +
                "    WATERMARK FOR eventtime AS eventtime - INTERVAL '5' SECOND,\n" +
                "    PRIMARY KEY (dt, user_id) NOT ENFORCED\n" +
                ") PARTITIONED BY (dt) \n" +
                ";";
        System.out.println("执行SQL语句：\n" + createTableSql);

        tableEnv.executeSql(createTableSql);

        // 打印SQL语句
        String insertSql = "INSERT OVERWRITE `paimon`.`default`.`user_behavior_copy` \n" +
                "SELECT user_id, item_id, behavior, dt, cast(eventtime as TIMESTAMP_LTZ(3))\n" +
                " FROM `paimon`.`default`.`user_behavior`\n";
        System.out.println("执行SQL语句：\n" + insertSql);

        // 使用INSERT OVERWRITE将数据从user_behavior表插入到user_behavior_copy表
        final TableResult tableResult = tableEnv.executeSql(insertSql);
        tableResult.await();

        // 打印SQL语句 - 更新所有behavior字段为0
        String updateSql = "UPDATE `paimon`.`default`.`user_behavior_copy` \n" +
                "SET behavior = 0";
        System.out.println("执行SQL语句：\n" + updateSql);
        
        // 执行UPDATE语句
        final TableResult tableResult1 = tableEnv.executeSql(updateSql);
        tableResult1.await();
    }
}